package com.zyh.study.myForestDemoPlus;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.f4b6a3.ulid.Ulid;
import com.zyh.study.myForestDemoPlus.OrderEventDTO;
import com.zyh.study.myForestDemoPlus.OrderSSEDeal;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBucket;
import org.redisson.api.RKeys;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @Classname EventCompensationService
 * @Description TODO()
 * @Date 2025-09-08 0:18
 * @Created by 86177
 */

@Slf4j
@Service
public class EventCompensationService {
    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private OrderSSEDeal orderSSEDeal;

    @Autowired
    private AlpacaSSEDeal alpacaSSEDeal;

    // JSON解析器（用于解析事件数据）
    private final ObjectMapper objectMapper = new ObjectMapper();
    private static final String MISSING_RANGE_PREFIX = "alpaca:missing:";
    private static final String UNPROCESSED_EVENT_PREFIX = "alpaca:event:";

    /**
     * 记录缺失的事件范围
     */
    public void recordMissingRange(String startId, String endId) {
        try {
            Ulid startUlid = Ulid.from(startId);
            Ulid endUlid = Ulid.from(endId);

            if (startUlid.compareTo(endUlid) > 0) {
                log.warn("Invalid range: {} > {}", startId, endId);
                return;
            }

            String rangeKey = MISSING_RANGE_PREFIX + startId + ":" + endId;
            RBucket<String> bucket = redissonClient.getBucket(rangeKey);
            bucket.set("pending", 24, TimeUnit.HOURS);
            log.info("Recorded missing range: {} - {}", startId, endId);
        } catch (Exception e) {
            log.error("Failed to record missing range: {}-{}", startId, endId, e);
        }
    }

    /**
     * 补偿缺失事件
     */
    public void compensateMissingEvents() {
        try {
            RKeys keys = redissonClient.getKeys();
            Iterable<String> rangeKeys = keys.getKeysByPattern(MISSING_RANGE_PREFIX + "*");

            for (String rangeKey : rangeKeys) {
                String[] ids = extractEventIds(rangeKey);
                if (ids == null || ids.length != 2) {
                    continue;
                }

                // 检查范围有效性
                if (isRangeStillValid(ids[0], ids[1])) {
                    fetchAndProcessRange(ids[0], ids[1]);
                }

                // 删除范围记录无论成功与否?
                redissonClient.getBucket(rangeKey).delete();
            }
        } catch (Exception e) {
            log.error("Compensation failed: {}", e.getMessage(), e);
        }
    }

    /**
     * 检查范围是否仍需要处理
     */
    private boolean isRangeStillValid(String startId, String endId) {
        try {
            RBucket<String> lastIdBucket = redissonClient.getBucket(LAST_PROCESSED_ID_KEY);
            Ulid lastUlid = Ulid.from(lastIdBucket.get());
            Ulid startUlid = Ulid.from(startId);

            // 如果当前最后处理ID已超过范围起始点，则范围无效
            return startUlid.compareTo(lastUlid) > 0;
        } catch (Exception e) {
            log.error("Range validation failed: {}-{}", startId, endId, e);
            return false;
        }
    }

    /**
     * 获取并处理事件范围
     */
    private void fetchAndProcessRange(String startId, String endId) {
        try {
            log.info("Fetching events from {} to {}", startId, endId);

            // 实际调用Alpaca API获取事件
            List<OrderEventDTO> events = orderSSEDeal.fetchEventRange(startId, endId);

            if (events == null || events.isEmpty()) {
                log.info("No events found in range {}-{}", startId, endId);
                return;
            }

            // 按事件ID排序处理
            events.sort(Comparator.comparing(OrderEventDTO::getUlid));

            for (OrderEventDTO event : events) {
                processCompensatedEvent(event);
            }

            log.info("Processed {} events in range {}-{}", events.size(), startId, endId);
        } catch (Exception e) {
            log.error("Failed to process range {}-{}", startId, endId, e);
        }
    }

    /**
     * 处理补偿事件（幂等保证）
     */
    private void processCompensatedEvent(OrderEventDTO event) {
        String eventKey = UNPROCESSED_EVENT_PREFIX + event.getEventId();
        RBucket<String> bucket = redissonClient.getBucket(eventKey);

        // 幂等检查：如果事件已存在且已处理，则跳过
        if (bucket.isExists()) {
            log.debug("Skipping already processed event: {}", event.getEventId());
            return;
        }

        try {
            // 保存事件
            bucket.set(objectMapper.writeValueAsString(event), 7, TimeUnit.DAYS);

            // 业务处理
            log.info("Processing compensated event: {}", event.getEventId());
            // orderService.processOrderEvent(event);

            // 更新最后处理ID
            alpacaSSEDeal.updateLastProcessedId(event.getEventId());

        } catch (Exception e) {
            log.error("Failed to process compensated event: {}", event.getEventId(), e);
        }
    }

    /**
     * 定时任务：每小时执行补偿
     */
    @Scheduled(fixedRate = 60 * 60 * 1000) // 每小时
    public void scheduledCompensation() {
        if (isSSEDisconnected()) {
            compensateMissingEvents();
        }
    }

    /**
     * 检查未处理事件
     */
    public void checkUnprocessedEvents() {
        try {
            RKeys keys = redissonClient.getKeys();
            Iterable<String> eventKeys = keys.getKeysByPattern(UNPROCESSED_EVENT_PREFIX + "*");

            int count = 0;
            for (String key : eventKeys) {
                RBucket<String> bucket = redissonClient.getBucket(key);
                if (bucket.isExists()) {
                    count++;
                    // 实际处理逻辑
                    // OrderEventDTO event = objectMapper.readValue(bucket.get(), OrderEventDTO.class);
                    // processCompensatedEvent(event);
                }
            }

            log.info("Found {} unprocessed events", count);
        } catch (Exception e) {
            log.error("Unprocessed event check failed: {}", e.getMessage(), e);
        }
    }

    /**
     * 从键名中提取事件ID
     */
    private String[] extractEventIds(String rangeKey) {
        try {
            String range = rangeKey.substring(MISSING_RANGE_PREFIX.length());
            String[] ids = range.split(":");
            if (ids.length == 2) {
                return new String[]{ids[0], ids[1]};
            }
        } catch (Exception e) {
            log.error("Invalid range key format: {}", rangeKey);
        }
        return null;
    }

    /**
     * 检查SSE连接状态
     */
    private boolean isSSEDisconnected() {
        RBucket<String> status = redissonClient.getBucket("alpaca:sse:status");
        return "disconnected".equals(status.get());
    }
}
